package com.xyy.saas.payment.util.mq;

import com.xyy.saas.payment.adpater.jd.common.JDConstants;
import com.xyy.saas.payment.common.RocketMqGroup;
import com.xyy.saas.payment.common.RocketMqTopic;
import com.xyy.saas.payment.common.constants.MqConstant;
import com.xyy.saas.payment.cores.vo.CallbackResultVo;
import com.xyy.saas.payment.cores.vo.ResultVO;
import com.xyy.saas.payment.cores.vo.jd.JDBankCardNotifyVo;
import com.xyy.saas.payment.util.CallbackUtil;
import com.xyy.saas.payment.util.JSONUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Map;


/**
 * @author wangtianqi
 */
@Service
@RocketMQMessageListener(topic = RocketMqTopic.TOPIC_RETRY_NOTIFY, consumerGroup = RocketMqGroup.GROUP_RETRY_NOTIFY, consumeMode = ConsumeMode.ORDERLY)
public class RetryNotifyConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(RetryNotifyConsumer.class);

    @Autowired
    MqProducer mqProducer;

    @Autowired
    CallbackUtil callbackUtil;

    @Override
    public void onMessage(MessageExt message) {
        String body = new String(message.getBody());
        try {
            LOGGER.info("RetryNotifyConsumer#onMessage body:{}", body);
            Map<String, Object> map = JSONUtils.parse2Map(body, String.class, Object.class);
            //取数据
            String url = (String) map.get("callBack");
            String count = (String) map.get("count");
            String result = (String) map.get("msg");
            //加次数
            int addedCount = Integer.parseInt(count) + 1;
            if (addedCount <= MqConstant.DELAY_LEVELS.length - 1) {
                //重发
                ResultVO resultVO = JSONUtils.parse2Object(result, ResultVO.class);
                if (JDConstants.CONTRACT_CALLBACK.equals(url)) {
                    JDBankCardNotifyVo jdBankCardNotifyVo = JSONUtils.parse2Object(resultVO.getResult().toString(), JDBankCardNotifyVo.class);
                    resultVO.setResult(jdBankCardNotifyVo);
                } else {
                    CallbackResultVo callbackResultVo = JSONUtils.parse2Object(resultVO.getResult().toString(), CallbackResultVo.class);
                    resultVO.setResult(callbackResultVo);
                }
                callbackUtil.sendRPCMsg(url, resultVO, addedCount);
            } else {
                //重试达到上限 记录日志
                LOGGER.error("RetryNotifyConsumer#onMessage call back count max error, url:{} count:{} result:{}", url, count, result);
            }

        } catch (Exception e) {
            LOGGER.error("RetryNotifyConsumer#onMessage consumer error body:{}", body, e);
        }
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis()));
    }

}

